package main

import (
	"bufio"
	"encoding/json"
	"flag"
	"fmt"
	"github.com/influxdata/influxdb1-client/v2"
	"io"
	"log"
	"net/http"
	"net/url"
	"os"
	"regexp"
	"strconv"
	"strings"

	//	"strings"
	"time"
)

type LogProcess struct{
	rc chan []byte
	wc chan *Message
	read Reader
	write Writer
}
//接口实现
type Reader interface {
	Read(rc chan []byte)
}
//接口
type Writer interface{
	Write(wc chan *Message)
}
type ReadFromFile struct{
	path string //读取文件路径

}
type WritetoDb struct{
	influxDBDsn string
}


type Message struct{
	TimeLocal time.Time
	BytesSent int
	Path ,Method,Scheme,Status string
	UpstreamTime,RequestTime float64
}
type SystemInfo struct{
	HandleLine int `json:"handleLine"` //总处理日志行数
	Tps float64 `json:"tps"` //系统吞吐量
	ReadChanLen int `json:"readChanLen"` //长度
	WriteChanLen int `json:"writeChanLen"` //长度
	RunTime string `json:"runTime"` //运行总时间
	ErrNum int `json:"errNum"` //错误数
}
type Monitor struct{
	startTime time.Time
	data SystemInfo
	tpsSli []int
}

//监控
const(
	//标记处理行数 读文件时
	TypeHandleLine =0
	TypeErrNum =1
)
var TypeMonitorChan = make(chan int ,200)
func (m * Monitor) start(lp *LogProcess){

	go func(){
		for n:=range TypeMonitorChan{
			switch n {
			case TypeErrNum:
				m.data.ErrNum +=1
			case TypeHandleLine:
				m.data.HandleLine+=1
			}
		}
	}()
	//定时器
	ticker :=time.NewTicker(time.Second*5)
	go func(){
		for{
			<-ticker.C
			//记录5秒处理量
			m.tpsSli =append(m.tpsSli,m.data.HandleLine)
			if len(m.tpsSli)>2{
				m.tpsSli=m.tpsSli[1:]
			}
		}
	}()
	http.HandleFunc("/monitor", func(writer http.ResponseWriter, request *http.Request) {
		//获取可读时间
		m.data.RunTime=time.Now().Sub(m.startTime).String()
		m.data.ReadChanLen =len(lp.rc)
		m.data.WriteChanLen =len (lp.wc)
		if len(m.tpsSli)>=2{
			m.data.Tps = float64((m.tpsSli[1]-m.tpsSli[0])/5)
		}
		ret,_:=json.MarshalIndent(m.data,"","\t")

		io.WriteString(writer,string(ret))
	})
	http.ListenAndServe(":8811",nil)
	os.Exit(1)

}
func (r *ReadFromFile) Read(rc chan []byte){
	//读取模块
	//line :="message"
	expath,_ := os.Getwd()
	fmt.Println(expath)

	f,err:=os.Open(r.path)

	if err!= nil{
		panic(fmt.Sprintf("open file error :%s",err.Error()))
	}
	//从文件末尾开始逐行读取文件内容
	f.Seek(0,2)
	rd := bufio.NewReader(f)
	for {
		line, err := rd.ReadBytes('\n')
		if err ==io.EOF{
			time.Sleep(1*time.Second)
			continue
		}else if err!=nil{
			panic (fmt.Sprintf("readbytes err %s\n",err.Error()))
		}
		//标记读取成功
		TypeMonitorChan<-TypeHandleLine
		//写入chan 去掉换行符
		rc <- line
	}
	//rc<-line

}


func (l *LogProcess) Process(){

	grepstr:=`^([\d\.]+)\s(-)\s(-)\s\[(.*?\s.*?)\]\s(.*?)\s(.*?)\s(.*?)\s(.*?)\s(.*?)\s(.*?)\s(.*?)\s`
	r:= regexp.MustCompile(grepstr)
	loc,_:=time.LoadLocation("Asia/Shanghai")
	//解析
	for v :=range l.rc {
		ret := r.FindStringSubmatch(string(v))

		message := &Message{}
		t, err := time.ParseInLocation("02/Jan/2006:15:04:05 +0800", ret[4], loc)
		if err != nil {
			//标记失败
			TypeMonitorChan<-TypeErrNum
			log.Println("parselocation fail", err.Error(), ret[4])
		}
		message.TimeLocal = t
		//字符串转int
		byteSent, _:=strconv.Atoi(ret[10])
		message.BytesSent=byteSent
		u,err:=url.Parse(ret[6])
		if(err !=nil){
			log.Println("url pares fail:",err)
			continue
		}
		message.Path=u.Path
		message.Method=ret[5]
		message.Status=ret[8]
		message.Scheme=ret[7]
		fmt.Print(message.Scheme,message.Path,message.Status,"-----",message.BytesSent,"-----",message.TimeLocal)
		l.wc<-message
	}
}

func(w *WritetoDb) Write(wc chan *Message){

	//循环读取chan 数据写入db
	//for v :=range wc{
	//
	//	fmt.Print(v)
	//}

	//os.Exit(1)
	// 写入模块

	infSli := strings.Split(w.influxDBDsn, "@")

	// Create a new HTTPClient
	c, err := client.NewHTTPClient(client.HTTPConfig{
		Addr:     infSli[0],
		Username: infSli[1],
		//Password: infSli[2],
	})
	if err != nil {
		log.Fatal(err)
	}

	for v := range wc {

		log.Println("write success!")
		fmt.Println(v)
		continue
		// Create a new point batch
		bp, err := client.NewBatchPoints(client.BatchPointsConfig{
			Database:  infSli[2],
			Precision: infSli[3],
		})
		if err != nil {
			log.Fatal(err)
		}

		// Create a point and add to batch
		// Tags: Path, Method, Scheme, Status
		tags := map[string]string{"Path": v.Path, "Method": v.Method, "Scheme": v.Scheme, "Status": v.Status}
		// Fields: UpstreamTime, RequestTime, BytesSent
		fields := map[string]interface{}{
			"UpstreamTime": v.UpstreamTime,
			"RequestTime":  v.RequestTime,
			"BytesSent":    v.BytesSent,
		}

		pt, err := client.NewPoint("accesslog", tags, fields, v.TimeLocal)
		if err != nil {
			log.Fatal(err)
		}
		bp.AddPoint(pt)

		// Write the batch
		if err := c.Write(bp); err != nil {
			log.Fatal(err)
		}

		log.Println("write success!")
	}
}
func main(){

	var path ,influxDsn string

	flag.StringVar(&path,"path","./access.log","read file patht")
	flag.StringVar(&influxDsn,"influxDsn","http://localhost:8086@admin@accesslog@s","influx")
	flag.Parse()
	fmt.Println(path,influxDsn)
	r:=&ReadFromFile{
		path:path,
	}
	w:=&WritetoDb{
		influxDBDsn:influxDsn,
	}

	//引用类型
	lp :=&LogProcess{
		rc:make(chan []byte,200),
		wc:make(chan *Message,200),
		read:r,
		write:w,
	}

	go lp.read.Read(lp.rc)

	for i:=0;i<2;i++ {
		go lp.Process()
	}
	for i:=0;i<4;i++ {
		go lp.write.Write(lp.wc)
	}
	m:= &Monitor{
		startTime:time.Now(),
		data:SystemInfo{},
	}
	m.start(lp)
	//time.Sleep(3000*time.Second)
}
